package com.mj.batch;

import com.mj.utils.Util;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 批量发送消息，只能同一个topic，且不是延迟消息或者事务消息
 */
public class BatchProducer {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 1.创建一个DefaultMQProducer实例，指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer(Util.ProducerGroup);
        // 2.设置NameServer的地址
        producer.setNamesrvAddr(Util.NamesrvAddr);
        // 3.启动生产者实例
        producer.start();
        // 4.批量消息，总大小不超过4MB，但是合理数据是1MB
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 150000; i++) {
            Message message = new Message(Util.Topic, "TagA", ("batch message" + i).getBytes(StandardCharsets.UTF_8));
            messages.add(message);
        }
        ListSplitter listSplitter = new ListSplitter(messages);
        int step = 0;
        while (listSplitter.hasNext()) {
            step++;
            // 5.批发送消息并接收发送结果
            SendResult sendResult = producer.send(listSplitter.next());
            // 打印发送结果，包括消息ID、发送状态等信息
            System.out.println(step + ":" + sendResult);
        }

        // 6.发送完所有消息后，关闭生产者实例，释放资源
        producer.shutdown();

    }
}

class ListSplitter implements Iterator<List<Message>> {
    private static final int SIZE_LIMIT = 10 * 1000;
    private final List<Message> messages;
    private int currentIndex;

    ListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    @Override
    public boolean hasNext() {
        return currentIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            //消息长度计算，body长度+topic长度
            int messageSize = message.getBody().length + message.getTopic().length();
            //消息长度计算，再加上配置信息长度
            Map<String, String> properties = message.getProperties();
            Iterator<Map.Entry<String, String>> iterator = properties.entrySet().iterator();
            while (iterator.hasNext()) {
                messageSize += iterator.next().getKey().length() + iterator.next().getValue().length();
            }
            //消息长度计算，再加20，估计可能还有一些其他信息大概20
            messageSize += 20;

            if (messageSize > SIZE_LIMIT) {
                //第一条信息就超出了限制，则直接发送第一条信息
                if (nextIndex - currentIndex == 0) {
                    nextIndex++;
                }
                break;
            }

            if (messageSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += messageSize;
            }
        }
        List<Message> messages = this.messages.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;
        return messages;
    }
}
